package com.ruoyi.project.flink;

import cn.hutool.json.JSONUtil;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.streaming.connectors.kafka.KafkaDeserializationSchema;
import org.apache.kafka.clients.consumer.ConsumerRecord;

import java.nio.charset.Charset;
import java.nio.charset.StandardCharsets;

public class UserInfoSchema implements KafkaDeserializationSchema<User> {

    public static final Charset UTF_8 = StandardCharsets.UTF_8;

    private static User buildMsg(String jsonString) {
        User user = JSONUtil.toBean(jsonString, User.class);
        return user;
    }

    @Override
    public boolean isEndOfStream(User user) {
        return false;
    }

    @Override
    public User deserialize(ConsumerRecord<byte[], byte[]> consumerRecord) throws Exception {
        String value = new String(consumerRecord.value(), UTF_8.name());
        return buildMsg(value);
    }

    @Override
    public TypeInformation<User> getProducedType() {
        return TypeInformation.of(User.class);
    }

}

